package com.gitee.melin.bee.core.fs.impl;

import com.gitee.melin.bee.core.fs.FileSystem;
import com.gitee.melin.bee.core.fs.FsConnectionInfo;
import java.util.List;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.awscore.exception.AwsServiceException;
import software.amazon.awssdk.http.HttpStatusCode;
import software.amazon.awssdk.http.apache.ApacheHttpClient;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.Bucket;
import software.amazon.awssdk.services.s3.model.ListBucketsResponse;

public class AwsS3FileSystem implements FileSystem {
    private static final Logger LOG = LoggerFactory.getLogger(AwsS3FileSystem.class);

    @Override
    public boolean bucketExist(FsConnectionInfo connectionInfo) {
        try (S3Client s3Client =
                S3Client.builder().region(Region.of(connectionInfo.getRegion())).build()) {

            s3Client.getBucketAcl(r -> r.bucket(connectionInfo.getBucket()));
            return true;
        } catch (AwsServiceException ase) {
            // https://docs.aws.amazon.com/code-library/latest/ug/s3_example_s3_Scenario_DoesBucketExist_section.html
            // A redirect error or an AccessDenied exception means the bucket exists but it's not in this region
            // or we don't have permissions to it.
            if ((ase.statusCode() == HttpStatusCode.MOVED_PERMANENTLY)
                    || "AccessDenied".equals(ase.awsErrorDetails().errorCode())) {
                return true;
            }
            if (ase.statusCode() == HttpStatusCode.NOT_FOUND) {
                return false;
            }
            throw ase;
        }
    }

    public void testConnection(FsConnectionInfo connectionInfo) {
        boolean exists = bucketExist(connectionInfo);
        if (!exists) {
            throw new IllegalArgumentException("bucket not exist: " + connectionInfo.getBucket());
        }
    }

    @Override
    public List<String> queryBuckets(FsConnectionInfo connectionInfo) {
        AwsCredentials awsCredentials =
                AwsBasicCredentials.create(connectionInfo.getAccessKeyId(), connectionInfo.getSecretAccessKey());
        StaticCredentialsProvider provider = StaticCredentialsProvider.create(awsCredentials);

        return queryBuckets(connectionInfo.getRegion(), provider);
    }

    private List<String> queryBuckets(String region, StaticCredentialsProvider provider) {
        try (S3Client s3Client = S3Client.builder()
                .httpClient(ApacheHttpClient.builder().build())
                .region(Region.of(region))
                .credentialsProvider(provider)
                .build()) {

            ListBucketsResponse bucketsResponse = s3Client.listBuckets();
            return bucketsResponse.buckets().stream().map(Bucket::name).collect(Collectors.toList());
        }
    }
}
